from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.metrics import accuracy_score
from sklearn.model_selection import KFold
import xgboost as xgb
from bayes_opt import BayesianOptimization
import joblib
from sklearn.svm import SVC
from sklearn.ensemble import RandomForestClassifier
import numpy as np
class Classifier:
"""
@brief Class representing a binary classifier with a XGBoost model.
@param validation (tuple): A tuple of validation datasets to be used for early stopping in XGBoost training.
"""
def __init__(self, validation):
"""
@brief Constructor method. Initializes main attributes.
"""
self.model = None
self.pipeline = None
self.best_params = None
self.validation = validation
def custom_cross_val(self, pipeline, X, y, n_splits=5, random_state=1):
"""
@brief Performs cross validation given a pipeline, features and targets.
@param pipeline: scikit-learn Pipeline object
@param X: numpy array or pandas DataFrame with features
@param y: numpy array or pandas Series with targets
@param n_splits: number of cross-validation folds
@param random_state: random seed for reproducibility
@return The mean of accuracy scores across all folds.
"""
# Define the cross-validator (here, we use n_splits-fold CV)
kf = KFold(n_splits=n_splits, random_state=random_state, shuffle=True)
# This will hold the cross-validation scores
cv_scores = []
# Perform the cross-validation manually
for train_index, val_index in kf.split(X):
cv_X_train, cv_X_val = X[train_index], X[val_index]
cv_y_train, cv_y_val = y[train_index], y[val_index]
if(self.model_type == "xgb"):
pipeline.fit(cv_X_train, cv_y_train,
classifier__eval_set=[(cv_X_val, cv_y_val)],
classifier__early_stopping_rounds=5, classifier__verbose=False)
else:
pipeline.fit(cv_X_train, cv_y_train)
y_pred = pipeline.predict(cv_X_val)
cv_scores.append(accuracy_score(cv_y_val, y_pred))
return np.mean(cv_scores)
def svm_func(self, C, gamma, X_train, y_train):
"""
Trains an SVM model with given hyperparameters and returns cross-validation score.
Params:
C: Penalty parameter C of the error term.
gamma: Kernel coefficient.
X_train: training features
y_train: training target
Returns: Cross-validation score.
"""
model = SVC(C=C, gamma=gamma, probability=True)
pipeline = Pipeline([('scaler', StandardScaler()),
('classifier', model)])
score = self.custom_cross_val(pipeline, X_train, y_train)
return score
def rf_func(self, n_estimators, max_depth, max_features, X_train, y_train):
"""
Trains a Random Forest model with given hyperparameters and returns cross-validation score.
Params:
n_estimators: The number of trees in the forest.
max_depth: The maximum depth of the tree.
max_features : The number of features to consider when looking for the best split
X_train: training features
y_train: training target
Returns: Cross-validation score.
"""
model = RandomForestClassifier(n_estimators=int(n_estimators),
max_depth=int(max_depth),
max_features=max_features,
random_state=1)
pipeline = Pipeline([('scaler', StandardScaler()),
('classifier', model)])
score = self.custom_cross_val(pipeline, X_train, y_train)
return score
def xgb_func(self, n_estimators, learning_rate, max_depth, subsample, colsample_bytree, gamma, X_train, y_train):
"""
@brief Trains a XGBoost model with given hyperparameters and returns cross-validation score.
@param n_estimators: number of boosted trees to fit
@param learning_rate: boosting learning rate
@param max_depth: maximum depth of a tree
@param subsample: subsample ratio of the training instances
@param colsample: subsample ratio of columns when constructing each tree
@param gamma: minimum loss reduction required to make a split
@param X_train: training features
@param y_train: training target
@return Cross-validation score.
"""
n_estimators = int(n_estimators)
max_depth = int(max_depth)
model = xgb.XGBClassifier(n_estimators=n_estimators,
learning_rate=learning_rate,
max_depth=max_depth,
subsample=subsample,
colsample_bytree=colsample_bytree,
gamma=gamma,
eval_metric='logloss',
objective='binary:logistic',
tree_method='gpu_hist', gpu_id=0)
pipeline = Pipeline([('scaler', StandardScaler()),
('classifier', model)])
score = self.custom_cross_val(pipeline, X_train, y_train)
return score
def optimize(self, X_train, y_train, model_type='xgb'):
"""
Uses Bayesian optimization to find the best hyperparameters for the model.
Params:
X_train: training features
y_train: training target
model_type: The type of model to use. Can be 'xgb', 'svm', or 'rf'.
"""
self.model_type = model_type
if model_type == 'xgb':
pbounds = {'learning_rate': (0.01, 0.2),
'n_estimators': (100, 1000),
'max_depth': (3, 7),
'subsample': (0.3, 0.7),
'colsample_bytree': (0.1, 0.5),
'gamma': (0, 5)}
optimizer = BayesianOptimization(
f=lambda n_estimators, learning_rate, max_depth, subsample, colsample_bytree, gamma: self.xgb_func(
n_estimators, learning_rate, max_depth, subsample, colsample_bytree, gamma, X_train, y_train),
pbounds=pbounds,
random_state=1,
)
elif model_type == 'svm':
pbounds = {'C': (0.1, 100),
'gamma': (0.0001, 1)}
optimizer = BayesianOptimization(
f=lambda C, gamma: self.svm_func(
C, gamma, X_train, y_train),
pbounds=pbounds,
random_state=1,
)
elif model_type == 'rf':
pbounds = {'n_estimators': (100, 1000),
'max_depth': (3, 7),
'max_features': (0.1, 0.5)}
optimizer = BayesianOptimization(
f=lambda n_estimators, max_depth, max_features: self.rf_func(
n_estimators, max_depth, max_features, X_train, y_train),
pbounds=pbounds,
random_state=1,
)
optimizer.maximize(init_points=3, n_iter=2)
self.best_params = optimizer.max['params']
def train(self, X_train, y_train, model_type='xgb'):
"""
Train the model with the best hyperparameters.
Params:
X_train: training features
y_train: training target
model_type: The type of model to use. Can be 'xgb', 'svm', or 'rf'.
Returns: Evaluated results of the model
"""
if model_type == 'xgb':
self.best_params['n_estimators'] = int(self.best_params['n_estimators'])
self.best_params['max_depth'] = int(self.best_params['max_depth'])
self.pipeline = Pipeline([('scaler', StandardScaler()),
('classifier', xgb.XGBClassifier(**self.best_params,
eval_metric='logloss',
objective='binary:logistic',
tree_method='gpu_hist', gpu_id=0))])
self.pipeline.fit(X_train, y_train,
classifier__eval_set=self.validation)
return self.pipeline.named_steps['classifier'].evals_result()
elif model_type == 'svm':
self.pipeline = Pipeline([('scaler', StandardScaler()),
('classifier', SVC(**self.best_params,probability=True))])
elif model_type == 'rf':
self.best_params['n_estimators'] = int(self.best_params['n_estimators'])
self.best_params['max_depth'] = int(self.best_params['max_depth'])
self.pipeline = Pipeline([('scaler', StandardScaler()),
('classifier', RandomForestClassifier(**self.best_params))])
self.pipeline.fit(X_train, y_train)
return None
def predict(self, X):
"""
@brief Predicts target with the trained model.
@param X: numpy array or pandas DataFrame with features
@return Predicted target values.
"""
return self.pipeline.predict(X)
def save_model(self, file_path):
"""
@brief Saves the trained model to a file.
@param file_path: path to the output file
"""
if self.pipeline is not None:
joblib.dump(self.pipeline, file_path)
else:
print("Cannot save an uninitialized model. Please train the model first.")
